<?php

namespace app\queue\stomp;

use Webman\Stomp\Client;
use app\queue\BaseConsumer;
use common\model\StepModel;
use support\bootstrap\Redis;
use common\service\StepService;
use Webman\Stomp\RetryAbleConsumer;
use malkusch\lock\mutex\PHPRedisMutex;

/**
 * 循环队列 执行写入sku数据
 */
class ExecuteGenerateStepSkuConsumer extends BaseConsumer implements RetryAbleConsumer
{

    // 要消费的队列名
    public $queue = 'execute_generate_step_sku';

    // 连接名，对应 config/redis_queue.php 里的连接`
    public $connection = 'default';

    /**
     * 消费
     * @param $data
     * @return mixed
     * @throws \Exception
     */
    public function doConsume($data)
    {
        $lockKey = $this->queue . '_:' . $data['step_id'];
        $mutex = new PHPRedisMutex([Redis::connection()->client()], $lockKey, $this->getTtr());
        return $mutex->synchronized(function () use ($data) {
            $redisDataKey = $data['data_key'];
            $batchData = [];
            $sep = $data['sep'];

            $step = model(StepModel::class)->where(['id' => $data['step_id']])->find();
            $attrList = StepService::getInstance()->getStepCategoryAttrList($step['step_category_id']);
            $attVariableDict = array_column($attrList, 'variable', 'value');

            $i = 0;
            while (Redis::lLen($redisDataKey)) {
                $i += 1;
                $sku = Redis::lPop($redisDataKey);
                $skuManHour = array_product(array_values(array_intersect_key($attVariableDict, array_flip(explode($sep, $sku)))));
                $manHour = number_floor_precision($step['variable'] * $skuManHour, 2);
                $batchData[] = [
                    'step_id' => $step['id'],
                    'uuid' => create_uuid(),
                    'attribute_value' => $sku,
                    'attribute_value_md5' => md5($sku),
                    'man_hour' => $manHour,
                ];
                if ($i >= $data['batch_number']) {
                    break;
                }
            }
            if (empty($batchData)) {
                return;
            }
            StepService::getInstance()->executeGenerateStepSku($batchData);
            if (Redis::lLen($redisDataKey)) {
                // 重设存活时间 避免没执行完就过期了
                Redis::expire($redisDataKey, 60 * 60 * 12);
                Client::send('execute_generate_step_sku', build_queue_data_with_xu($data), 1);
            }
        });
    }

    public function getTtr()
    {
        return 30;
    }

    public function canRetry($attempt, $error)
    {
        trace(format_exception_info($error), 'ERR');
        return $attempt <= 3;
    }
}
